#include <stdio.h>
#include "redis.h"
#include "z_logger.h"

static REDIS redis = NULL;

_Bool redis_open(const char *host) {
    if (NULL == redis) {
        redis = credis_connect(host, 0, 1000);
        if (NULL == redis) {
            fprintf(stderr, "redis server connect failed");
        }
    }

    return NULL != redis;
}

void redis_close() {
    if (NULL != redis) {
        credis_close(redis);
        redis = NULL;
    }
}

int redis_set_value(const device_t *device, const thing_model_t *tag, const value_t *value) {
    int rc;

    char key[64], val[128];
    sprintf(key, "%d_%d_%d", device->super.id, tag->segment, tag->super.id);
    value_printf(val, value);

    rc = credis_set(redis, key, val);
    if (CREDIS_OK != rc) {
        log_error("credis_set=%d", rc);
    }

    return rc;
}

int redis_get_values(const device_t *device, const vector_t *tag_list, vector_t *val_list) {
    int i;
    int rc;

    if (0 == vector_size(tag_list)) {
        return 0;
    }

    vector_t key_list = vector_initializer;

    for (i = 0; i < vector_size(tag_list); ++i) {
        char *key = vector_item_new(32);
        thing_model_t *tag = vector_get(tag_list, i);
        sprintf(key, "%d_%d_%d", device->super.id, tag->segment, tag->super.id);
        vector_add(&key_list, key);
    }

    char **val = NULL;
    rc = credis_mget(redis, vector_size(&key_list), (const char **) key_list.vals, &val);
    vector_foreach_free(&key_list);
    vector_clear(&key_list);

    if (rc < 0) {
        log_error("credis_mget=%d", rc);
    }

    for (i = 0; i < vector_size(tag_list) && i < rc; ++i) {
        value_t *value = value_new();
        value_scanf(val[i], value);
        vector_add(val_list, value);
    }

    return rc;
}

int redis_pub_value(const channel_t *channel, const device_t *device, const thing_model_t *tag, const value_t *value) {
    int rc;

    char key[64], val[128];
    sprintf(key, "%d_%d_%d", device->super.id, tag->segment, tag->super.id);
    value_printf(val, value);

    char topic[64], message[256];
    sprintf(topic, "pub.kv.%d.%d.%d", channel->super.id, device->super.id, tag->segment);
    sprintf(message, "%s={%s}", key, val);

    rc = credis_publish(redis, topic, message);
    if (CREDIS_OK != rc) {
        log_error("credis_publish=%d", rc);
    }

    return rc;
}

int redis_sub_value(value_id_t *id, value_t *value) {
    int rc;
    char *pattern, *topic, *message;

    rc = credis_listen(redis, &pattern, &topic, &message);
    if (CREDIS_OK != rc) {
        log_error("credis_listen=%d", rc);
        return rc;
    }

    rc = sscanf(topic, "pub.kv.%d.%d.%d", &id->chn_id, &id->dev_id, &id->seg_id);
    if (rc < 3) { return -1; }

    char key[64], val[128];
    rc = sscanf(message, "%s={%s}", key, val);
    if (rc < 2) { return -1; }

    rc = sscanf(key, "%d_%d_%d", &id->dev_id, &id->seg_id, &id->tag_id);
    if (rc < 3) { return -1; }

    value_scanf(val, value);

    return 0;
}

int redis_pub_bytes(const char *topic, const uint8_t *ptr, size_t nbytes) {
    char hex[nbytes * 2 + 1];
    size_t len = sizeof(hex);

    bin2str(ptr, nbytes, hex, &len);
    return credis_publish(redis, topic, hex);
}

int redis_sub_bytes(char **topic, uint8_t *ptr, size_t *nbytes) {
    int rc;
    char *pattern, *message;

    rc = credis_listen(redis, &pattern, topic, &message);
    if (CREDIS_OK != rc) {
        log_error("credis_listen=%d", rc);
        return rc;
    }

    return NULL != str2bin(message, ptr, nbytes) ? CREDIS_OK : CREDIS_ERR_NOMEM;
}

int redis_subscribe(const char *topic) {
    return credis_subscribe(redis, topic);
}